package com.test.chat.server;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.test.chat.handler.MqttHandler;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.AdaptiveRecvByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.epoll.EpollChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.mqtt.MqttDecoder;
import io.netty.handler.codec.mqtt.MqttEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;

public class NettyServer {

	private static final Logger LOGGER = LoggerFactory.getLogger(NettyServer.class);

	private int port;
	private int readerIdleTimeSeconds = 0;
	private int writerIdleTimeSeconds = 0;
	private int allIdleTimeSeconds = 100;

	public NettyServer(int port) {
		this.port = port;
	}

	public void start() {
		NioEventLoopGroup bossGroup = new NioEventLoopGroup();
		NioEventLoopGroup workerGroup = new NioEventLoopGroup();
		try {
			ServerBootstrap b = new ServerBootstrap();
			b.group(bossGroup, workerGroup)
			.channel(NioServerSocketChannel.class)
			.childHandler(new ChannelInitializer<Channel>() {
				@Override
				protected void initChannel(Channel ch) throws Exception {
					ChannelPipeline pipeline = ch.pipeline();
					pipeline.addLast("logingHandler", new LoggingHandler(LogLevel.DEBUG));
					pipeline.addLast("mqttDecoder", new MqttDecoder());
					pipeline.addLast("mqttEncoder", new MqttEncoder());
					pipeline.addLast("idleStateHandler", new IdleStateHandler(readerIdleTimeSeconds,
							writerIdleTimeSeconds, allIdleTimeSeconds));
					pipeline.addLast("mqttHandler", new MqttHandler());
				}
			})
			.option(ChannelOption.SO_BACKLOG, 128)
            .option(ChannelOption.SO_REUSEADDR, true)
            .option(ChannelOption.SO_RCVBUF, 10 * 1024)
            .option(ChannelOption.SO_SNDBUF, 10 * 1024)
            .option(EpollChannelOption.SO_REUSEPORT, true)
            .childOption(ChannelOption.SO_KEEPALIVE, true)
            .option(ChannelOption.TCP_NODELAY, true)
            .childOption(ChannelOption.RCVBUF_ALLOCATOR, AdaptiveRecvByteBufAllocator.DEFAULT)
            .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
            .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
			
			LOGGER.info(b.toString());
			ChannelFuture f = b.bind(port).sync();
			LOGGER.info("server started, bind port:{}", port);
			
			f.channel().closeFuture().sync();
		} catch (InterruptedException e) {
			e.printStackTrace();
			LOGGER.error(e.getMessage());
		} finally {
			workerGroup.shutdownGracefully();
			bossGroup.shutdownGracefully();
			LOGGER.info("server closed");
		}
	}

}
